package com.mars.kafkademo.controller;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * author by: mars
 * Date: 2020/4/12 17:39
 * Description: 消费者
 */
public class CustomConsumer {

    public static void main(String[] args) {
        Properties props = new Properties();
        // kafka服务端的ip:port 可替换成 主机名:port
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.5.115:9092");
        // 制定消费者组
        props.put(ConsumerConfig.GROUP_ID_CONFIG,"aa");
        // 是否自动确认offset
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");
        // 自动确认offset的时间间隔
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000");
        // key反序列化，注意消费者的反序列化使用包 "org.apache.kafka.common.serialization.StringDeserializer"
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // value反序列化
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(props);
        consumer.subscribe(Arrays.asList("mars11"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record:records) {
                System.out.printf("offset=%d,key=%s,value=%s%n",record.offset(),record.key(),record.value());
            }
        }
    }

}
